// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <sstream>
#include <string>
#include <vector>

#include <nebula/array/builder_base.h>
#include <nebula/array/builder_binary.h>
#include <nebula/array/builder_nested.h>
#include <nebula/compute/api_vector.h>
#include <nebula/compute/function.h>
#include <nebula/compute/type_fwd.h>

#include <nebula/types/scalar.h>
#include <turbo/utility/status.h>
#include <turbo/base/checked_cast.h>
#include <nebula/util/key_value_metadata.h>
#include <nebula/util/reflection_internal.h>
#include <nebula/util/string.h>
#include <turbo/base/macros.h>

namespace nebula {
    struct Scalar;
    struct StructScalar;

    namespace internal {
        template<>
        struct EnumTraits<compute::SortOrder>
                : BasicEnumTraits<compute::SortOrder, compute::SortOrder::Ascending,
                        compute::SortOrder::Descending> {
            static std::string name() { return "SortOrder"; }

            static std::string value_name(compute::SortOrder value) {
                switch (value) {
                    case compute::SortOrder::Ascending:
                        return "Ascending";
                    case compute::SortOrder::Descending:
                        return "Descending";
                }
                return "<INVALID>";
            }
        };
    }  // namespace internal

    namespace compute {
        namespace internal {

            using nebula::internal::EnumTraits;
            using nebula::internal::has_enum_traits;

            template<typename Enum, typename CType = typename std::underlying_type<Enum>::type>
            turbo::Result<Enum> ValidateEnumValue(CType raw) {
                for (auto valid: EnumTraits<Enum>::values()) {
                    if (raw == static_cast<CType>(valid)) {
                        return static_cast<Enum>(raw);
                    }
                }
                return turbo::invalid_argument_error("Invalid value for ", EnumTraits<Enum>::name(), ": ", raw);
            }

            class TURBO_EXPORT GenericOptionsType : public FunctionOptionsType {
            public:
                turbo::Result<std::shared_ptr<Buffer>> serialize(const FunctionOptions &) const override;

                turbo::Result<std::unique_ptr<FunctionOptions>> deserialize(BufferSpan buffer) const override;

                virtual turbo::Status ToStructScalar(const FunctionOptions &options,
                                              std::vector<std::string> *field_names,
                                              std::vector<std::shared_ptr<Scalar>> *values) const = 0;

                virtual turbo::Result<std::unique_ptr<FunctionOptions>> FromStructScalar(
                        const StructScalar &scalar) const = 0;
            };

            TURBO_EXPORT
            turbo::Result<std::shared_ptr<StructScalar>> FunctionOptionsToStructScalar(
                    const FunctionOptions &);

            TURBO_EXPORT
            turbo::Result<std::unique_ptr<FunctionOptions>> FunctionOptionsFromStructScalar(
                    const StructScalar &);

            TURBO_EXPORT
            turbo::Result<std::unique_ptr<FunctionOptions>> DeserializeFunctionOptions(BufferSpan buffer);

            template<typename T>
            static inline enable_if_t<!has_enum_traits<T>::value, std::string> GenericToString(
                    const T &value) {
                std::stringstream ss;
                ss << value;
                return ss.str();
            }

            template<typename T>
            static inline enable_if_t<!has_enum_traits<T>::value, std::string> GenericToString(
                    const std::optional<T> &value) {
                return value.has_value() ? GenericToString(value.value()) : "nullopt";
            }

            static inline std::string GenericToString(bool value) { return value ? "true" : "false"; }

            static inline std::string GenericToString(const std::string &value) {
                std::stringstream ss;
                ss << '"' << value << '"';
                return ss.str();
            }

            template<typename T>
            static inline enable_if_t<has_enum_traits<T>::value, std::string> GenericToString(
                    const T value) {
                return EnumTraits<T>::value_name(value);
            }

            template<typename T>
            static inline std::string GenericToString(const std::shared_ptr<T> &value) {
                std::stringstream ss;
                return value ? value->to_string() : "<nullptr>";
            }

            static inline std::string GenericToString(const std::shared_ptr<Scalar> &value) {
                std::stringstream ss;
                if (value) {
                    ss << value->type->to_string() << ":" << value->to_string();
                } else {
                    ss << "<nullptr>";
                }
                return ss.str();
            }

            static inline std::string GenericToString(
                    const std::shared_ptr<const KeyValueMetadata> &value) {
                std::stringstream ss;
                ss << "KeyValueMetadata{";
                if (value) {
                    bool first = true;
                    for (const auto &pair: value->sorted_pairs()) {
                        if (!first) ss << ", ";
                        first = false;
                        ss << pair.first << ':' << pair.second;
                    }
                }
                ss << '}';
                return ss.str();
            }

            static inline std::string GenericToString(const Datum &value) {
                switch (value.kind()) {
                    case Datum::NONE:
                        return "<NULL DATUM>";
                    case Datum::SCALAR:
                        return GenericToString(value.scalar());
                    case Datum::ARRAY: {
                        std::stringstream ss;
                        ss << value.type()->to_string() << ':' << value.make_array()->to_string();
                        return ss.str();
                    }
                    default:
                        return value.to_string();
                }
            }

            template<typename T>
            static inline std::string GenericToString(const std::vector<T> &value) {
                std::stringstream ss;
                ss << "[";
                bool first = true;
                // Don't use range-for with auto& to avoid Clang -Wrange-loop-analysis
                for (auto it = value.begin(); it != value.end(); it++) {
                    if (!first) ss << ", ";
                    first = false;
                    ss << GenericToString(*it);
                }
                ss << ']';
                return ss.str();
            }

            static inline std::string GenericToString(SortOrder value) {
                switch (value) {
                    case SortOrder::Ascending:
                        return "Ascending";
                    case SortOrder::Descending:
                        return "Descending";
                }
                return "<INVALID SORT ORDER>";
            }

            static inline std::string GenericToString(const std::vector<SortKey> &value) {
                std::stringstream ss;
                ss << '[';
                bool first = true;
                for (const auto &key: value) {
                    if (!first) {
                        ss << ", ";
                    }
                    first = false;
                    ss << key.to_string();
                }
                ss << ']';
                return ss.str();
            }

            template<typename T>
            static inline bool GenericEquals(const T &left, const T &right) {
                return left == right;
            }

            template<typename T>
            static inline bool GenericEquals(const std::shared_ptr<T> &left,
                                             const std::shared_ptr<T> &right) {
                if (left && right) {
                    return left->equals(*right);
                }
                return left == right;
            }

            static inline bool IsEmpty(const std::shared_ptr<const KeyValueMetadata> &meta) {
                return !meta || meta->size() == 0;
            }

            static inline bool GenericEquals(const std::shared_ptr<const KeyValueMetadata> &left,
                                             const std::shared_ptr<const KeyValueMetadata> &right) {
                // Special case since null metadata is considered equivalent to empty
                if (IsEmpty(left) || IsEmpty(right)) {
                    return IsEmpty(left) && IsEmpty(right);
                }
                return left->equals(*right);
            }

            template<typename T>
            static inline bool GenericEquals(const std::vector<T> &left,
                                             const std::vector<T> &right) {
                if (left.size() != right.size()) return false;
                for (size_t i = 0; i < left.size(); i++) {
                    if (!GenericEquals(left[i], right[i])) return false;
                }
                return true;
            }

            template<typename T>
            static inline decltype(TypeTraits<typename CTypeTraits<T>::ArrowType>::type_singleton())
            GenericTypeSingleton() {
                return TypeTraits<typename CTypeTraits<T>::ArrowType>::type_singleton();
            }

            template<typename T>
            static inline enable_if_same<T, std::shared_ptr<const KeyValueMetadata>,
                    std::shared_ptr<DataType>>
            GenericTypeSingleton() {
                return map(binary(), binary());
            }

            template<typename T>
            static inline enable_if_t<has_enum_traits<T>::value, std::shared_ptr<DataType>>
            GenericTypeSingleton() {
                return TypeTraits<typename EnumTraits<T>::Type>::type_singleton();
            }

            template<typename T>
            static inline enable_if_same<T, SortKey, std::shared_ptr<DataType>>
            GenericTypeSingleton() {
                std::vector<std::shared_ptr<Field>> fields;
                fields.emplace_back(new Field("target", GenericTypeSingleton<std::string>()));
                fields.emplace_back(new Field("order", GenericTypeSingleton<SortOrder>()));
                return std::make_shared<StructType>(std::move(fields));
            }

// N.B. ordering of overloads is relatively fragile
            template<typename T>
            static inline turbo::Result<decltype(MakeScalar(std::declval<T>()))> GenericToScalar(
                    const T &value) {
                return MakeScalar(value);
            }

// For Clang/libc++: when iterating through vector<bool>, we can't
// pass it by reference so the overload above doesn't apply
            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(bool value) {
                return MakeScalar(value);
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(const FieldRef &ref) {
                return MakeScalar(ref.to_dot_path());
            }

            template<typename T, typename Enable = enable_if_t<has_enum_traits<T>::value>>
            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(const T value) {
                using CType = typename EnumTraits<T>::CType;
                return GenericToScalar(static_cast<CType>(value));
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(const SortKey &key) {
                TURBO_MOVE_OR_RAISE(auto target, GenericToScalar(key.target));
                TURBO_MOVE_OR_RAISE(auto order, GenericToScalar(key.order));
                return StructScalar::create({target, order}, {"target", "order"});
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(
                    const std::shared_ptr<const KeyValueMetadata> &value) {
                auto ty = GenericTypeSingleton<std::shared_ptr<const KeyValueMetadata>>();
                std::unique_ptr<ArrayBuilder> builder;
                TURBO_RETURN_NOT_OK(MakeBuilder(default_memory_pool(), ty, &builder));
                auto *map_builder = turbo::checked_cast<MapBuilder *>(builder.get());
                auto *key_builder = turbo::checked_cast<BinaryBuilder *>(map_builder->key_builder());
                auto *item_builder = turbo::checked_cast<BinaryBuilder *>(map_builder->item_builder());
                TURBO_RETURN_NOT_OK(map_builder->append());
                if (value) {
                    TURBO_RETURN_NOT_OK(key_builder->append_values(value->keys()));
                    TURBO_RETURN_NOT_OK(item_builder->append_values(value->values()));
                }
                std::shared_ptr<Array> arr;
                TURBO_RETURN_NOT_OK(map_builder->finish(&arr));
                return arr->get_scalar(0);
            }

            template<typename T>
            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(
                    const std::vector<T> &value) {
                std::shared_ptr<DataType> type = GenericTypeSingleton<T>();
                std::vector<std::shared_ptr<Scalar>> scalars;
                scalars.reserve(value.size());
                // Don't use range-for with auto& to avoid Clang -Wrange-loop-analysis
                for (auto it = value.begin(); it != value.end(); it++) {
                    TURBO_MOVE_OR_RAISE(auto scalar, GenericToScalar(*it));
                    scalars.push_back(std::move(scalar));
                }
                std::unique_ptr<ArrayBuilder> builder;
                TURBO_RETURN_NOT_OK(
                        MakeBuilder(default_memory_pool(), type ? type : scalars[0]->type, &builder));
                TURBO_RETURN_NOT_OK(builder->append_scalars(scalars));
                std::shared_ptr<Array> out;
                TURBO_RETURN_NOT_OK(builder->finish(&out));
                return std::make_shared<ListScalar>(std::move(out));
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(
                    const std::shared_ptr<DataType> &value) {
                if (!value) {
                    return turbo::invalid_argument_error("shared_ptr<DataType> is nullptr");
                }
                return MakeNullScalar(value);
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(const TypeHolder &value) {
                return GenericToScalar(value.get_shared_ptr());
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(
                    const std::shared_ptr<Scalar> &value) {
                return value;
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(
                    const std::shared_ptr<Array> &value) {
                return std::make_shared<ListScalar>(value);
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(const Datum &value) {
                // TODO(ARROW-9434): store in a union instead.
                switch (value.kind()) {
                    case Datum::ARRAY:
                        return GenericToScalar(value.make_array());
                        break;
                    default:
                        return turbo::unimplemented_error("Cannot serialize Datum kind ", value.kind());
                }
            }

            static inline turbo::Result<std::shared_ptr<Scalar>> GenericToScalar(std::nullopt_t) {
                return std::make_shared<NullScalar>();
            }

            template<typename T>
            static inline auto GenericToScalar(const std::optional<T> &value)
            -> turbo::Result<decltype(MakeScalar(value.value()))> {
                return value.has_value() ? MakeScalar(value.value()) : std::make_shared<NullScalar>();
            }

            template<typename T>
            static inline enable_if_primitive_ctype<typename CTypeTraits<T>::ArrowType, turbo::Result<T>>
            GenericFromScalar(const std::shared_ptr<Scalar> &value) {
                using ArrowType = typename CTypeTraits<T>::ArrowType;
                using ScalarType = typename TypeTraits<ArrowType>::ScalarType;
                if (value->type->id() != ArrowType::type_id) {
                    return turbo::invalid_argument_error("Expected type ", ArrowType::type_id, " but got ",
                                           value->type->to_string());
                }
                const auto &holder = turbo::checked_cast<const ScalarType &>(*value);
                if (!holder.is_valid) return turbo::invalid_argument_error("Got null scalar");
                return holder.value;
            }

            template<typename T>
            static inline enable_if_primitive_ctype<typename EnumTraits<T>::Type, turbo::Result<T>>
            GenericFromScalar(const std::shared_ptr<Scalar> &value) {
                TURBO_MOVE_OR_RAISE(auto raw_val,
                                      GenericFromScalar<typename EnumTraits<T>::CType>(value));
                return ValidateEnumValue<T>(raw_val);
            }

            template<typename T, typename U>
            using enable_if_same_result = enable_if_same<T, U, turbo::Result<T>>;

            template<typename T>
            static inline enable_if_same_result<T, std::string> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                if (!is_base_binary_like(value->type->id())) {
                    return turbo::invalid_argument_error("Expected binary-like type but got ", value->type->to_string());
                }
                const auto &holder = turbo::checked_cast<const BaseBinaryScalar &>(*value);
                if (!holder.is_valid) return turbo::invalid_argument_error("Got null scalar");
                return holder.value->to_string();
            }

            template<typename T>
            static inline enable_if_same_result<T, FieldRef> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                TURBO_MOVE_OR_RAISE(auto path, GenericFromScalar<std::string>(value));
                return FieldRef::from_dot_path(path);
            }

            template<typename T>
            static inline enable_if_same_result<T, SortKey> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                if (value->type->id() != Type::STRUCT) {
                    return turbo::invalid_argument_error("Expected type STRUCT but got ", value->type->id());
                }
                if (!value->is_valid) return turbo::invalid_argument_error("Got null scalar");
                const auto &holder = turbo::checked_cast<const StructScalar &>(*value);
                TURBO_MOVE_OR_RAISE(auto target_holder, holder.field("target"));
                TURBO_MOVE_OR_RAISE(auto order_holder, holder.field("order"));
                TURBO_MOVE_OR_RAISE(auto target, GenericFromScalar<FieldRef>(target_holder));
                TURBO_MOVE_OR_RAISE(auto order, GenericFromScalar<SortOrder>(order_holder));
                return SortKey{std::move(target), order};
            }

            template<typename T>
            static inline enable_if_same_result<T, std::shared_ptr<DataType>> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                return value->type;
            }

            template<typename T>
            static inline enable_if_same_result<T, TypeHolder> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                return value->type;
            }

            template<typename T>
            static inline enable_if_same_result<T, std::shared_ptr<Scalar>> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                return value;
            }

            template<typename T>
            static inline enable_if_same_result<T, std::shared_ptr<const KeyValueMetadata>>
            GenericFromScalar(const std::shared_ptr<Scalar> &value) {
                auto ty = GenericTypeSingleton<std::shared_ptr<const KeyValueMetadata>>();
                if (!value->type->equals(ty)) {
                    return turbo::invalid_argument_error("Expected ", ty->to_string(), " but got ",
                                           value->type->to_string());
                }
                const auto &holder = turbo::checked_cast<const MapScalar &>(*value);
                std::vector<std::string> keys;
                std::vector<std::string> values;
                const auto &list = turbo::checked_cast<const StructArray &>(*holder.value);
                const auto &key_arr = turbo::checked_cast<const BinaryArray &>(*list.field(0));
                const auto &value_arr = turbo::checked_cast<const BinaryArray &>(*list.field(1));
                for (int64_t i = 0; i < list.length(); i++) {
                    keys.push_back(key_arr.get_string(i));
                    values.push_back(value_arr.get_string(i));
                }
                return key_value_metadata(std::move(keys), std::move(values));
            }

            template<typename T>
            static inline enable_if_same_result<T, Datum> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                if (value->type->id() == Type::LIST) {
                    const auto &holder = turbo::checked_cast<const BaseListScalar &>(*value);
                    return holder.value;
                }
                // TODO(ARROW-9434): handle other possible datum kinds by looking for a union
                return turbo::invalid_argument_error("Cannot deserialize Datum from ", value->to_string());
            }

            template<typename>
            constexpr inline bool is_optional_v = false;
            template<typename T>
            constexpr inline bool is_optional_v<std::optional<T>> = true;
            template<>
            constexpr inline bool is_optional_v<std::nullopt_t> = true;

            template<typename T>
            static inline std::enable_if_t<is_optional_v<T>, turbo::Result<T>> GenericFromScalar(
                    const std::shared_ptr<Scalar> &value) {
                using value_type = typename T::value_type;
                if (value->type->id() == Type::NA) {
                    return std::nullopt;
                }
                return GenericFromScalar<value_type>(value);
            }

            template<typename T>
            static enable_if_same<typename CTypeTraits<T>::ArrowType, ListType, turbo::Result<T>>
            GenericFromScalar(const std::shared_ptr<Scalar> &value) {
                using VT = typename T::value_type;
                if (value->type->id() != Type::LIST) {
                    return turbo::invalid_argument_error("Expected type LIST but got ", value->type->to_string());
                }
                const auto &holder = turbo::checked_cast<const BaseListScalar &>(*value);
                if (!holder.is_valid) return turbo::invalid_argument_error("Got null scalar");
                std::vector<VT> result;
                for (int i = 0; i < holder.value->length(); i++) {
                    TURBO_MOVE_OR_RAISE(auto scalar, holder.value->get_scalar(i));
                    TURBO_MOVE_OR_RAISE(auto v, GenericFromScalar<VT>(scalar));
                    result.push_back(std::move(v));
                }
                return result;
            }

            template<typename Options>
            struct StringifyImpl {
                template<typename Tuple>
                StringifyImpl(const Options &obj, const Tuple &props)
                        : obj_(obj), members_(props.size()) {
                    props.ForEach(*this);
                }

                template<typename Property>
                void operator()(const Property &prop, size_t i) {
                    std::stringstream ss;
                    ss << prop.name() << '=' << GenericToString(prop.get(obj_));
                    members_[i] = ss.str();
                }

                std::string finish() {
                    return "{" + nebula::internal::JoinStrings(members_, ", ") + "}";
                }

                const Options &obj_;
                std::vector<std::string> members_;
            };

            template<typename Options>
            struct CompareImpl {
                template<typename Tuple>
                CompareImpl(const Options &l, const Options &r, const Tuple &props)
                        : left_(l), right_(r) {
                    props.ForEach(*this);
                }

                template<typename Property>
                void operator()(const Property &prop, size_t) {
                    equal_ &= GenericEquals(prop.get(left_), prop.get(right_));
                }

                const Options &left_;
                const Options &right_;
                bool equal_ = true;
            };

            template<typename Options>
            struct ToStructScalarImpl {
                template<typename Tuple>
                ToStructScalarImpl(const Options &obj, const Tuple &props,
                                   std::vector<std::string> *field_names,
                                   std::vector<std::shared_ptr<Scalar>> *values)
                        : obj_(obj), field_names_(field_names), values_(values) {
                    props.ForEach(*this);
                }

                template<typename Property>
                void operator()(const Property &prop, size_t) {
                    if (!status_.ok()) return;
                    auto result = GenericToScalar(prop.get(obj_));
                    if (!result.ok()) {
                        status_ = result.status().with_message("Could not serialize field ", prop.name(),
                                                              " of options type ", Options::kTypeName, ": ",
                                                              result.status().message());
                        return;
                    }
                    field_names_->emplace_back(prop.name());
                    values_->push_back(result.move_value_unsafe());
                }

                const Options &obj_;
                turbo::Status status_;
                std::vector<std::string> *field_names_;
                std::vector<std::shared_ptr<Scalar>> *values_;
            };

            template<typename Options>
            struct FromStructScalarImpl {
                template<typename Tuple>
                FromStructScalarImpl(Options *obj, const StructScalar &scalar, const Tuple &props)
                        : obj_(obj), scalar_(scalar) {
                    props.ForEach(*this);
                }

                template<typename Property>
                void operator()(const Property &prop, size_t) {
                    if (!status_.ok()) return;
                    auto maybe_holder = scalar_.field(std::string(prop.name()));
                    if (!maybe_holder.ok()) {
                        status_ = maybe_holder.status().with_message(
                                "Cannot deserialize field ", prop.name(), " of options type ",
                                Options::kTypeName, ": ", maybe_holder.status().message());
                        return;
                    }
                    auto holder = maybe_holder.move_value_unsafe();
                    auto result = GenericFromScalar<typename Property::Type>(holder);
                    if (!result.ok()) {
                        status_ = result.status().with_message("Cannot deserialize field ", prop.name(),
                                                              " of options type ", Options::kTypeName, ": ",
                                                              result.status().message());
                        return;
                    }
                    prop.set(obj_, result.move_value_unsafe());
                }

                Options *obj_;
                turbo::Status status_;
                const StructScalar &scalar_;
            };

            template<typename Options>
            struct CopyImpl {
                template<typename Tuple>
                CopyImpl(Options *obj, const Options &options, const Tuple &props)
                        : obj_(obj), options_(options) {
                    props.ForEach(*this);
                }

                template<typename Property>
                void operator()(const Property &prop, size_t) {
                    prop.set(obj_, prop.get(options_));
                }

                Options *obj_;
                const Options &options_;
            };

            template<typename Options, typename... Properties>
            const FunctionOptionsType *get_function_options_type(const Properties &... properties) {
                static const class OptionsType : public GenericOptionsType {
                public:
                    explicit OptionsType(const nebula::internal::PropertyTuple<Properties...> properties)
                            : properties_(properties) {}

                    const char *type_name() const override { return Options::kTypeName; }

                    std::string stringify(const FunctionOptions &options) const override {
                        const auto &self = turbo::checked_cast<const Options &>(options);
                        return StringifyImpl<Options>(self, properties_).finish();
                    }

                    bool compare(const FunctionOptions &options,
                                 const FunctionOptions &other) const override {
                        const auto &lhs = turbo::checked_cast<const Options &>(options);
                        const auto &rhs = turbo::checked_cast<const Options &>(other);
                        return CompareImpl<Options>(lhs, rhs, properties_).equal_;
                    }

                    turbo::Status ToStructScalar(const FunctionOptions &options,
                                          std::vector<std::string> *field_names,
                                          std::vector<std::shared_ptr<Scalar>> *values) const override {
                        const auto &self = turbo::checked_cast<const Options &>(options);
                        TURBO_RETURN_NOT_OK(
                                ToStructScalarImpl<Options>(self, properties_, field_names, values).status_);
                        return turbo::OkStatus();
                    }

                    turbo::Result<std::unique_ptr<FunctionOptions>> FromStructScalar(
                            const StructScalar &scalar) const override {
                        auto options = std::make_unique<Options>();
                        TURBO_RETURN_NOT_OK(
                                FromStructScalarImpl<Options>(options.get(), scalar, properties_).status_);
                        // R build with openSUSE155 requires an explicit unique_ptr construction
                        return std::unique_ptr<FunctionOptions>(std::move(options));
                    }

                    std::unique_ptr<FunctionOptions> copy(const FunctionOptions &options) const override {
                        auto out = std::make_unique<Options>();
                        CopyImpl<Options>(out.get(), turbo::checked_cast<const Options &>(options), properties_);
                        return out;
                    }

                private:
                    const nebula::internal::PropertyTuple<Properties...> properties_;
                } instance(nebula::internal::MakeProperties(properties...));
                return &instance;
            }

            turbo::Status CheckAllArrayOrScalar(const std::vector<Datum> &values);

            TURBO_EXPORT
            turbo::Result<std::vector<TypeHolder>> GetFunctionArgumentTypes(const std::vector<Datum> &args);

        }  // namespace internal
    }  // namespace compute
}  // namespace nebula
